package ar.edu.itba.ontologias.TPONTO;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapred.JobConf;

import bixo.datum.FetchedDatum;
import bixo.datum.ParsedDatum;
import cascading.scheme.SequenceFile;
import cascading.scheme.TextLine;
import cascading.tap.Hfs;
import cascading.tap.Tap;
import cascading.tuple.TupleEntry;
import cascading.tuple.TupleEntryIterator;

public class App {

	static List<Person> personas = new ArrayList<Person>();
	static List<Movie> peliculas = new ArrayList<Movie>();

	@SuppressWarnings("unused")
	public static void main(String[] args) throws IOException {
		final String curWorkingDirPath = Constantes.fileName;
		// final Tap crawldbTap = new Hfs(new SequenceFile(CrawlDbDatum.FIELDS),
		// new Path(curWorkingDirPath, "crawldb").toString());
		final Tap contentTap = new Hfs(new SequenceFile(FetchedDatum.FIELDS),
				new Path(curWorkingDirPath, "content").toString());
		final Tap parseTap = new Hfs(new SequenceFile(ParsedDatum.FIELDS),
				new Path(curWorkingDirPath, "parse").toString());
		final Tap statusTap = new Hfs(new TextLine(), new Path(
				curWorkingDirPath, "status").toString());
		final JobConf conf = new JobConf();
		final TupleEntryIterator it = contentTap.openForRead(conf);
		Set<String> set = new HashSet<String>();
		List<Critica> criticas = new ArrayList<Critica>();
		int i = 0;
		while (it.hasNext()) {
			final TupleEntry t = it.next();
			final FetchedDatum datum = new FetchedDatum(t);
			String type = Helper.getType(datum.getFetchedUrl());
			try{
				switch (Type.valueOf(type.toUpperCase())) {
					case PERSONA:
						// System.out.println(new String(datum.getContentBytes()));
						Person persona = Helper.getPerson(new String(datum
								.getContentBytes()));
						//personas.add(persona);
						Helper.makeRDF(persona);
						//System.out.println("Persona = " + persona);
						i++;
						break;
					case PELICULA:
						// System.out.println(new String(datum.getContentBytes()));
						Movie pelicula = Helper.getMovie(new String(datum
								.getContentBytes()));
						peliculas.add(pelicula);
						//System.out.println("Pelicula = " + pelicula);
						i++;
						Helper.makeRDF(pelicula);
						break;
					case CRITICA:
						// Critica critica = Helper.getCritica(new
						// String(datum.getContentBytes()));
						// System.out.println("Critica");
						break;
					default:
						break;
					}
			}catch(Exception e){
				//System.out.println(type);
			}
			// System.out.println(new String(datum.getFetchedUrl()));
			// set.add(Helper.getType(datum.getFetchedUrl()));
			// Helper.getAlgo(new String(datum.getContentBytes()));
		}
		
	}

	public static List<Person> getPersonas() {
		return personas;
	}

}